package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    // 阻塞队列，存放秒杀订单信息  修改后不再使用
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
    // 订单处理线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    // 构造方法 > @Autowired > @PostConstruct
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    // 直接调用 redis 消息队列信息
    private class VoucherOrderHandler implements Runnable {
        String queueName = "stream.orders";

        // run 方法不断从阻塞队列里取
        @Override
        public void run() {
            while (true) {
                // 获取队类中的订单信息
                try {
                    // 获取消息队列中的订单信息  XREADGROUP g1 c1 COUNT 1 Block 2000 STREAMS streams.order >
                    List<MapRecord<String, Object, Object>> list=
                            // stringRedisTemplate.opsForStream().read(...)：这是 Spring Data Redis 提供的 API，
                            // 用于从 Redis Stream 中读取消息。
                            stringRedisTemplate.opsForStream().read(
                                    // onsumer.from("g1", "c1")：指定消费者组和消费者名称。
                                    // g1 是消费者组名称，c1 是消费者名称。
                            Consumer.from("g1", "c1"),
                            // StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2))：
                           //count(1)：每次最多读取一条消息。
                            //block(Duration.ofSeconds(2))：如果当前没有消息，阻塞等待 2 秒
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            // StreamOffset.create(queueName, ReadOffset.lastConsumed())：
                                    // 指定从哪个位置开始读取消息。queueName 是 Redis Stream 的名称，
                           // ReadOffset.lastConsumed() 表示从上次消费的位置开始读取。
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );
                    // 判断消息队列获取是否成工
                    if (list==null || list.isEmpty()){
                       // 如果获取失败 说明没有消息 继续下一次循环
                        continue;
                    }
                    // 解析消息中的订单消息
                    // list.get(0)：获取读取到的第一条消息。
                    //record.getValue()：获取消息的内容，返回一个 Map<Object, Object>，
                    // 其中键是消息字段，值是字段对应的值。
                    //BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true)：
                    // 使用工具类 BeanUtil 将消息内容填充到 VoucherOrder 对象中。true 表示忽略大小写。
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 如果获取失败 说明没有消息 继续下一次
                    //如果获取成功 可以下单
                    handleVoucherOrder(voucherOrder);
                    // ACK 确定 移出
                   stringRedisTemplate.opsForStream().acknowledge("queueName","g1",record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                    handlePendingList();
                }

            }
        }



        private void handlePendingList() {
            while (true) {
                // 获取队类中的订单信息
                try {
                    // 获取消息队列中的订单信息  XREADGROUP g1 c1 COUNT 1 Block 2000 STREAMS streams.order >
                    List<MapRecord<String, Object, Object>> list= stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.from("0"))
                    );
                    // 判断消息队列获取是否成工
                    if (list==null || list.isEmpty()){
                        // 如果获取失败 说明没有消息 继续下一次循环
                        break;
                    }
                    // 解析消息中的订单消息
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 如果获取失败 说明没有消息 继续下一次
                    //如果获取成功 可以下单
                    handleVoucherOrder(voucherOrder);
                    // ACK 确定 移出
                    stringRedisTemplate.opsForStream().acknowledge("queueName","g1",record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                }

            }
        }




    }





    // 用户可能在秒杀开始启动就直接开始抢购，所以这里需要提前初始化一个订单
    // 开启第二个线程实现异步下单
    /*private class VoucherOrderHandler implements Runnable{
        // run 方法不断从阻塞队列里取
        @Override
        public void run() {
            while (true){
                // 获取队类中的订单信息
                try {
                    VoucherOrder voucherOrder = orderTasks.take();
                    // 获取阻塞队列头部 没有元素阻塞
                    // 创建订单
                    handleVoucherOrder(voucherOrder);
                } catch (InterruptedException e) {
                    log.error("处理订单异常",e);
                }

            }
        }
    }*/
// 1
    // 处理订单逻辑 创建订单
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        RLock lock = redissonClient.getLock("order" + voucherOrder.getUserId());
        boolean isLock = lock.tryLock(); //锁获比重试最大等待时间 无参 -1 失败不等待
        if (!isLock) {
            // 获取锁失败 返回错误或重试
            log.error("获取订单锁失败");
            return;
        }
        try {
// 此时进到 2
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            lock.unlock();
            //simpleRedisLock.unlock();
        }

    }

    // lua 脚本静态加载
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    // 注入 Redisson 框架客户端的RedissonClient
    @Resource
    private RedissonClient redissonClient;


    // 改造使用redis消息队列后的
    @Override
    public Result setkillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");
        // 执行  lua 脚本
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(),
                userId.toString(),
                String.valueOf(voucherId)
        );
        // 判断结果是否为 0
        int r = result.intValue();
        if (r != 0) {
            // 秒杀失败
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 返回订单 id
        return Result.ok(orderId);

    }


    // 这是秒杀业务 检查时间是否过期 以及库存是否足够
/*    @Override
//    @Transactional
    public Result setkillVoucher(Long voucherId) {
        // 查询  需要注入优惠卷service
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 秒杀是否开始
        if(voucher.getBeginTime().isAfter(LocalDateTime.now())){
            // 秒杀尚未开始
            return Result.fail("秒杀尚未开始");
        };
        // 判断秒杀是否结束
        if(voucher.getEndTime().isBefore(LocalDateTime.now())){
            // 秒杀已经结束
            return Result.fail("秒杀已经结束");
        }
        // 判断是否充足库存
        if(voucher.getStock()<1){
            return Result.fail("库存不足");
        }
        // 扣减库存
        Long userId = UserHolder.getUser().getId();
        //SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order"+userId,stringRedisTemplate);
        RLock lock = redissonClient.getLock("order" + userId);
        //boolean isLock = simpleRedisLock.tryLock(1200);
        boolean isLock = lock.tryLock();  //锁获取是啊比重试最大等待时间 无参 -1 失败不等待
        if (!isLock){
            // 获取锁失败 返回错误或重试
            return Result.fail("您已下过一单订单，禁止重复下单");
        }
        try{
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }finally {
            lock.unlock();
            //simpleRedisLock.unlock();
        }
        */

    private IVoucherOrderService proxy;
// 任何人进来先调用 setkillVoucher 再 1 2







       /* @Override
        public Result setkillVoucher(Long voucherId) {
            // 执行  lua 脚本
            Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
                    Collections.emptyList(),
                    voucherId.toString(),
                    UserHolder.getUser().getId().toString()
            );
            // 判断结果是否为 0
            int r= result.intValue();
            if(r!=0){
                // 秒杀失败
                return Result.fail(r==1?"库存不足":"不能重复下单");
            }
            //  不是 0 代表没有购买资格
            // 为 0 有购买资格 把下单信息保存到订单队列
            // 创建订单 不是直接下单 将这个扔到阻塞队列
            VoucherOrder voucherOrder = new VoucherOrder();
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);
            // 用户id
            Long userId = UserHolder.getUser().getId();
            voucherOrder.setUserId(userId);
            voucherOrder.setVoucherId(voucherId);
            // 放到阻塞队列 然后执行异步下单
            orderTasks.add(voucherOrder);
            // 获取代理对象
            // 使用代理对象解决 Spring 事务失效场景问题
            proxy = (IVoucherOrderService) AopContext.currentProxy();
            // 返回订单 id
            return Result.ok(orderId);
        }*/


    // 不在使用 synchronized 不能解决集群部署一致性
        /*synchronized (userId.toString().intern()){
           //  return createVoucherOrder(voucherId);
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }*/
// 2
    @Transactional
    public Result createVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        // 一人一单
        int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder).count();
        if (count > 0) {
            log.error("您已下过一单订单，禁止重复下单");
            return null;
        }
        boolean success = seckillVoucherService.update()
                .setSql("stock=stock-1")
                .eq("voucher_id", voucherOrder)
//                .eq("stock",voucher.getStock())
                .gt("stock", 0)  // 构建条件查询器 大于 0
                .update();
        // 创建订单
        if (!success) {
            // 扣减失败
            log.error("扣减库存失败");
            return null;
        }
        save(voucherOrder);
        return null;
    }

}
